use std::{sync::atomic::{AtomicBool, Ordering}, thread};
use std::sync::mpsc::{self, Receiver, Sender, RecvTimeoutError};
use std::time::{Duration, Instant};
use crate::error::ServerError;
use std::sync::Arc;
use zmq;

const HTTP_INTERVAL: u64 = 2000;
const CONNECTION_TIMEOUT_SECS: u64 = 10; // 10秒连接超时

pub fn get_pipe_handle(pipe_addr: String) -> Result<(Receiver<String>, Arc<AtomicBool>), ServerError> {
    let (tx, rx) = mpsc::channel();
    let (conn_tx, conn_rx) = mpsc::channel();
    let flag: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
    let flag_http = flag.clone();
    let flag_tcp = flag.clone();
    match pipe_addr {
        addr if addr.starts_with("http://") => {
            let pipe_addr_clone = addr.clone();
            let handle = thread::spawn(move || {
                let rt = tokio::runtime::Runtime::new().unwrap();
                rt.block_on(async move {
                    let client = reqwest::Client::new();
                    let start_time = Instant::now();
                    let mut first_success = false;
                    loop {
                        if !flag_http.load(Ordering::Relaxed) {
                            break;
                        }
                        match client.get(&pipe_addr_clone)
                        .timeout(Duration::from_secs(3))
                        .send().await {
                            Ok(resp) => {
                                if let Ok(text) = resp.text().await {
                                    if !first_success {
                                        first_success = true;
                                        let _ = conn_tx.send(Ok(()));
                                    }
                                    let _ = tx.send(text);
                                }
                            }
                            Err(e) => {
                                if !first_success && start_time.elapsed().as_secs() > CONNECTION_TIMEOUT_SECS {
                                    let _ = conn_tx.send(Err(ServerError::TaskErrorStr("连接超时")));
                                }
                                break;
                            }
                        }
                        tokio::time::sleep(Duration::from_millis(500)).await;
                    }
                    let _ = tx.send("Closed".to_string());
                });
            });
        },
        addr if addr.starts_with("tcp://") => {
            todo!()
        }
        _ => return Err(ServerError::TaskErrorStr("无效的管道地址"))
    };
    // 试听
    match conn_rx.recv_timeout(Duration::from_secs(CONNECTION_TIMEOUT_SECS)) {
        Ok(result) => {
            match result {
                Ok(_) => Ok((rx, flag.clone())),
                Err(e) => Err(e),
            }
        },
        Err(RecvTimeoutError::Timeout) => {
            println!("tasks - stream - get_pipe_handle - 连接超时，未收到连接状态");
            Err(ServerError::TaskErrorStr("连接超时，未收到连接状态"))
        },
        Err(RecvTimeoutError::Disconnected) => {
            println!("tasks - stream - get_pipe_handle - 通道已关闭");
            Err(ServerError::TaskErrorStr("通道已关闭"))
        }
    }
}